package org.hashdb.client.net.nio;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.hashdb.client.console.AppConsoleReader;
import org.hashdb.client.net.nio.msg.ActMessage;
import org.hashdb.client.net.nio.msg.ErrorMessage;
import org.hashdb.client.net.nio.msg.Message;
import org.hashdb.client.util.AsyncService;

import java.util.LinkedHashMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

/**
 * Date: 2024/1/21 10:44
 *
 * @author huanyuMake-pecdle
 * @version 0.0.1
 */
public class ActMessageHandler extends ChannelInboundHandlerAdapter {

    private Channel channel;

    private final AppConsoleReader reader;

    private final LinkedHashMap<Long, RequestBuilder<? extends Message<?>>> responseMap = new LinkedHashMap<>();

    private final LinkedHashMap<Long, Consumer<Message<?>>> responseConsumerMap = new LinkedHashMap<>();

    public ActMessageHandler(AppConsoleReader reader) {
        this.reader = reader;
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        channel = ctx.channel();
    }

    public class ResponseConsumerBuilder {
        private Consumer<Message<?>> consumer;

    }

    public class ResponseConsumer implements Consumer<Consumer<Message<?>>> {

        private Consumer<Message<?>> consumer;

        @Override
        public void accept(Consumer<Message<?>> consumer) {
            this.consumer = consumer;
        }

        private void consume(Message<?> m) {
            consumer.accept(m);
        }
    }

    private void doResponseMapChannelRead(ChannelHandlerContext ctx, Object msg) {
        if (!(msg instanceof Message<?>)) {
            ctx.fireChannelRead(msg);
            return;
        }
        if (!(msg instanceof ActMessage<?> act)) {
            ctx.fireChannelRead(msg);
            return;
        }
        var wrapper = responseMap.remove(act.actId());
        if (wrapper != null) {
            wrapper.responseFuture.complete(act);
            if (!wrapper.intercept()) {
                ctx.fireChannelRead(msg);
            }
            return;
        }
        if (!(msg instanceof ErrorMessage errorMessage)) {
            reader.errorLn("can not found matched request of response '" + msg + "'");
            return;
        }
        var firstRequest = responseMap.firstEntry();
        if (firstRequest == null) {
            ctx.fireChannelRead(msg);
            return;
        }
        firstRequest.getValue().responseFuture.complete(errorMessage);
    }

    /*private void doResponseConsumerMapChannelRead(ChannelHandlerContext ctx, Object msg) {
        if (!(msg instanceof Message<?>)) {
            ctx.fireChannelRead(msg);
            return;
        }
        if (!(msg instanceof ActMessage<?> act)) {
            ctx.fireChannelRead(msg);
            return;
        }
        var responseConsumer = responseConsumerMap.remove(act.actId());
        if (responseConsumer != null) {
            responseConsumer.accept(act);
            if (!wrapper.intercept()) {
                ctx.fireChannelRead(msg);
            }
            return;
        }
        if (!(msg instanceof ErrorMessage errorMessage)) {
            reader.errorLn("can not found matched request of response '" + msg + "'");
            return;
        }
        var firstRequest = responseMap.firstEntry();
        if (firstRequest == null) {
            ctx.fireChannelRead(msg);
            return;
        }
        firstRequest.getValue().responseFuture.complete(errorMessage);
    }
*/
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        doResponseMapChannelRead(ctx, msg);
    }

    public <Req extends Message<?>> RequestBuilder<Req> newRequest(Req request) {
        return new RequestBuilder<>(request);
    }

    class RequestBuilder<Req extends Message<?>> {

        private final Req request;
        private final CompletableFuture<Message<?>> responseFuture = new CompletableFuture<>();

        /**
         * ms
         */
        private long timeout = 5_000;

        /**
         * 拦截, 是否在 {@link ActMessageHandler} 的读事件方法中匹配了相应结果后
         * 继续将结果交由 后面处理器 处理
         */
        private boolean intercept = true;

        private ScheduledFuture<?> timeoutTask;

        public RequestBuilder(Req request) {
            this.request = request;
            responseFuture.thenApplyAsync(m -> {
                timeoutTask.cancel(true);
                clear();
                return m;
            }, AsyncService.service());
        }

        private void doRequest() {
            reader.messageLn("do request " + request);
            responseMap.put(request.id(), this);
            channel.writeAndFlush(request);
        }

        private void clear() {
            responseMap.remove(request.id());
        }

        public RequestBuilder<Req> shouldIntercept() {
            this.intercept = true;
            return this;
        }

        public boolean intercept() {
            return intercept;
        }

        public RequestBuilder<Req> timeout(long timeout) {
            this.timeout = timeout;
            return this;
        }

        public CompletableFuture<Message<?>> apply() throws CompletionException {
            doRequest();
            timeoutTask = AsyncService.setTimeout(() -> {
                responseFuture.completeExceptionally(new TimeoutException());
                clear();
            }, timeout);
            return responseFuture;
        }

        void cancelTimeout() {
            if (timeoutTask != null) {
                timeoutTask.cancel(true);
            }
        }
    }

}
